package task

import (
	"context"
	"fmt"
	"gitee.com/yijimu/gin-frame/utils"
	"github.com/go-redsync/redsync/v4"
	"github.com/hibiken/asynq"
	"github.com/spf13/viper"
	"log"
	"time"
)

// Server 延迟队列(任务)、分布式任务、定时任务 及 计划任务(周期性的定时任务) 服务端
type Server struct {
	TaskHandler      []Handler                        // 任务处理方法集合
	CronTaskListFunc func() ([]CronTaskConfig, error) //计划(周期)任务列表
}

// Run 启动 处理任务 及 计划任务 相关服务
func (server *Server) Run() {
	// 创建任务处理器服务 asynq.Server
	srv := asynq.NewServer(
		asynq.RedisClientOpt{
			Addr:     viper.GetString("task.redis.addr"),
			Password: viper.GetString("task.redis.password"),
			DB:       viper.GetInt("task.redis.db"),
		},
		asynq.Config{ // Config用于调整服务器的任务处理行为
			Concurrency: 10, // 表示最大并发处理任务数。
		},
	)

	mux := asynq.NewServeMux()
	for _, taskHandler := range server.TaskHandler {
		mux.HandleFunc(taskHandler.Type, taskHandler.Handler)
	}

	// 系统预设周期计划任务
	mux.HandleFunc("cron:ping", func(ctx context.Context, task *asynq.Task) error {
		return nil
	})

	go func() {
		// 解决 asynq 调度器 在多台服务器部署时，计划任务 重复
		// 当有一台服务器 开启了计划任务(根据cronIsOpen来判断)，其他服务器就不允许开启
		// 当已开启计划任务的服务器下线后，其余服务器接管
		for {
			// 使用分布式锁来保证只有一台服务器开启计划任务服务
			if redisSync, err := utils.InitRedsync(viper.GetString("task.redis.addr"), viper.GetString("task.redis.password"), viper.GetInt("task.redis.db")); err == nil {
				if err := redisSync.Handler("asynq-cron-mutex", func() error {
					// 动态添加和删除周期性任务（即不重新启动Scheduler进程），可以使用PeriodicTaskManager。PeriodTaskManager使用PeriodTaskConfigProvider定期获取当前定期任务配置，并将调度器的条目与当前配置同步。
					// 可以将周期性任务配置存储在数据库或本地文件中，并更新此配置源以动态添加和删除周期性任务
					if mgr, err := asynq.NewPeriodicTaskManager(asynq.PeriodicTaskManagerOpts{
						RedisConnOpt: asynq.RedisClientOpt{
							Addr:     viper.GetString("task.redis.addr"),
							Password: viper.GetString("task.redis.password"),
							DB:       viper.GetInt("task.redis.db"),
						},
						PeriodicTaskConfigProvider: server,           // 配置源的接口
						SyncInterval:               10 * time.Second, // 指定同步发生的频率（同步配置源）
					}); err != nil {
						fmt.Println(err.Error())
					} else {
						// 运行调度器 并 非阻塞
						fmt.Println("计划周期任务服务开启了~")
						if err := mgr.Run(); err != nil {
							log.Fatal("计划周期任务服务启动失败：" + err.Error())
						}
					}
					return nil
				}, true, []redsync.Option{
					redsync.WithTries(10),                    //获取锁次数
					redsync.WithRetryDelay(10 * time.Second), // 每次 获取锁 的等待时间
					redsync.WithExpiry(8 * time.Second),
				}...); err != nil {
					fmt.Println(err.Error())
				}
			} else {
				fmt.Println(err.Error())
			}
			time.Sleep(10 * time.Second)
		}
	}()

	if err := srv.Run(mux); err != nil {
		log.Fatal(err)
	}
}

// Handler 处理任务方法
type Handler struct {
	Type    string                                            // 任务名称(标识)
	Handler func(ctx context.Context, task *asynq.Task) error // 任务处理函数
}

func (server *Server) GetConfigs() ([]*asynq.PeriodicTaskConfig, error) { //实现的方法接口：PeriodicTaskConfigProvider  读取所有配置项
	configs := []*asynq.PeriodicTaskConfig{
		{ // 系统预留周期计划任务
			Cronspec: "@every 24h",
			Task:     asynq.NewTask("cron:ping", []byte("{ \"desc\": \"系统预留周期任务，不得删除及停止\" }")),
		},
	}

	if cronTaskList, err := server.CronTaskListFunc(); err == nil {
		for _, cfg := range cronTaskList {
			configs = append(configs, &asynq.PeriodicTaskConfig{
				Cronspec: cfg.Cronspec,
				Task:     asynq.NewTask(cfg.TaskPattern, []byte(cfg.TaskParameter)),
			})
		}
	}

	return configs, nil
}

type CronTaskConfig struct {
	Cronspec      string `gorm:"column:cronspec;type:varchar(255);not null;default:''" json:"cronspec"`
	TaskPattern   string `gorm:"column:task_pattern;type:varchar(255);not null;default:''" json:"task_pattern"`
	TaskParameter string `gorm:"column:task_parameter;type:varchar(255);not null;default:''" json:"task_parameter"`
}
